
#include <queue>

#include <boost/thread/mutex.hpp>
#include <boost/thread/condition_variable.hpp>

#include "table_schema_manager.hpp"
#include "table_manager.hpp"
#include "service_provider.hpp"
#include "kv_scanner_manager.hpp"

#include "global.hpp"

namespace xbase
{
	namespace global
	{
		boost::shared_ptr<TabletServerConfiguration> the_conf;

		boost::shared_ptr<kv_scanner_manager> the_scanner_manager;
		boost::shared_ptr<table_schema_manager> the_schema_manager;
		boost::shared_ptr<table_manager> the_table_manager;

		boost::shared_ptr<protocol::server_address> the_address_info;
		boost::shared_ptr<ServerLoad> the_load_info;

		boost::shared_ptr<root_meta_manager> the_root_meta_manager;
		log_manager::pointer the_log_manager;

		boost::shared_ptr<service_provider> the_service;
		boost::shared_ptr<master_session> the_master_session;

		fs::path the_tablet_server_path;

		bool stop_request;

		void init() {
			stop_request = false;
			the_conf.reset(new TabletServerConfiguration());

			the_scanner_manager.reset(new kv_scanner_manager());
			the_schema_manager.reset(new table_schema_manager());
			the_table_manager.reset(new table_manager());

			the_address_info.reset(new protocol::server_address());
			the_load_info.reset(new ServerLoad());

			the_root_meta_manager.reset(new root_meta_manager());
			the_log_manager = log_manager::pointer(new log_manager());

			the_service.reset(new service_provider());
			the_master_session.reset(new master_session());
		}
		
		void reset() {
			stop_minor_compact();
			stop_major_compact();

			// delete others by order
			the_service.reset();
			the_master_session.reset();

			the_scanner_manager.reset();
			the_schema_manager.reset();
			the_table_manager.reset();
			the_root_meta_manager.reset();

			the_address_info.reset();
			the_load_info.reset();

			// 最后将日志线程停掉，它只在处理完所有日志以后才会退出。
			global::the_log_manager->stop();
			global::the_log_manager->join();

			// 在这里再删除日志，因为tablet中的group析构时用到了它
			the_log_manager.reset();

			the_conf.reset();
			stop_request = false;
		}

		// minor compact section

		static
		std::queue<xbase::tablet::pointer> minor_compact_tablet_list;
		static
		bool minor_compact_stopped = false;
		static
		boost::mutex minor_compact_tablet_list_lock;
		static
		boost::condition_variable minor_compact_tablet_list_not_empty_condition;

		void notify_minor_compact(tablet::pointer tablet)
		{
			LOG(TRACE, "notify_minor_compact tablet " << tablet->tablet_name());
			{
				boost::mutex::scoped_lock lock(minor_compact_tablet_list_lock);
				if(! minor_compact_stopped)
				{
					minor_compact_tablet_list.push(tablet);
				}
			}
			minor_compact_tablet_list_not_empty_condition.notify_one();
		}

		tablet::pointer get_minor_compact_tablet(boost::system_time t)
		{
			boost::mutex::scoped_lock lock(minor_compact_tablet_list_lock);
			while(minor_compact_tablet_list.empty())
			{
				const bool is_not_timeout = minor_compact_tablet_list_not_empty_condition.timed_wait(lock, t);
				if(minor_compact_stopped)
				{
					return tablet::pointer();
					break; // is stopped
				}
				if(! is_not_timeout)
				{
					break; // is timeout
				}
			}
			// 等待超时，或有子表加入进来了
			tablet::pointer p;
			if(! minor_compact_tablet_list.empty())
			{
				p = minor_compact_tablet_list.front();
				minor_compact_tablet_list.pop();
			}
			return p;
		}

		// 不要直接调用它，而是调用相应checker中的stop
		void stop_minor_compact() {
			{
				boost::mutex::scoped_lock lock(minor_compact_tablet_list_lock);
				while(! minor_compact_tablet_list.empty())
				{
					minor_compact_tablet_list.pop();
				}
				minor_compact_stopped = true;
			}
			minor_compact_tablet_list_not_empty_condition.notify_all();
		}


		// major compact section

		static
		std::queue<xbase::tablet::pointer> major_compact_tablet_list;
		static
		bool major_compact_stopped = false;
		static
		boost::mutex major_compact_tablet_list_lock;
		static
		boost::condition_variable major_compact_tablet_list_not_empty_condition;

		void notify_major_compact(tablet::pointer tablet)
		{
			LOG(TRACE, "notify_major_compact tablet " << tablet->tablet_name());
			{
				boost::mutex::scoped_lock lock(major_compact_tablet_list_lock);
				if(! major_compact_stopped)
				{
					major_compact_tablet_list.push(tablet);
				}
			}
			major_compact_tablet_list_not_empty_condition.notify_one();
		}

		tablet::pointer get_major_compact_tablet(boost::system_time t)
		{
			boost::mutex::scoped_lock lock(major_compact_tablet_list_lock);
			while(major_compact_tablet_list.empty())
			{
				const bool is_not_timeout = major_compact_tablet_list_not_empty_condition.timed_wait(lock, t);
				if(minor_compact_stopped)
				{
					return tablet::pointer();
					break; // is stopped
				}
				if(! is_not_timeout)
				{
					break; // is timeout
				}
			}
			// 等待超时，或有子表加入进来了
			tablet::pointer p;
			if(! major_compact_tablet_list.empty())
			{
				p = major_compact_tablet_list.front();
				major_compact_tablet_list.pop();
			}
			return p;
		}

		// 不要直接调用它，而是调用相应checker中的stop
		void stop_major_compact() {
			{
				boost::mutex::scoped_lock lock(major_compact_tablet_list_lock);
				while(! major_compact_tablet_list.empty())
				{
					major_compact_tablet_list.pop();
				}
				major_compact_stopped = true;
			}
			major_compact_tablet_list_not_empty_condition.notify_all();
		}

	} // namespace global
	
} // namespace xbase
	
